package org.yuanzheng.tableAndSql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.yuanzheng.source.StationLog

/*批计算*/
object TestCreateTableByDataStream {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //读取数据
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.10", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })


    /*方式一，有表名称*/
    //tableEnv.registerDataStream("t_station_log2", stream) // 注册一张表
    //val table1: Table = tableEnv.scan("t_station_log2") //打印表结构，或者使用table api，需要获得Table对象
    //table1.printSchema()
    //tableEnv.sqlQuery("select * from t_station_log2").printSchema()  //打印表结构

    /*方式二，没有表名称*/
    //val table2: Table = tableEnv.fromDataStream(stream, 'id, 'call_out, 'call_in)  //基于位置修改字段名称
    val table2: Table = tableEnv.fromDataStream(stream,
      'sid as 'station_id, 'callOut as 'call_out, 'callIn as 'call_in, 'callType as 'call_type, 'callTime as 'call_time, 'duration as 'duration_time)
    table2.printSchema()

    streamEnv.execute()
  }
}
